feat(java): add mesh scheduling worker#359
Conversation
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 26 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (4)
📝 WalkthroughWalkthroughThis PR adds an optional "mesh" clustering feature to taskito-java, gating a new ChangesMesh scheduling feature
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant JavaWorker as Java Worker
participant JNI as NativeWorker (JNI)
participant SpawnMesh as spawn_mesh
participant Bridge as run_mesh_bridge
participant MeshNode
JavaWorker->>JNI: start(options with meshConfig)
JNI->>SpawnMesh: spawn_mesh(config_json, worker_id, queues)
SpawnMesh->>MeshNode: construct + start gossip/steal-server
SpawnMesh->>Bridge: spawn run_mesh_bridge
loop job processing
Bridge->>MeshNode: drain local deque
MeshNode-->>Bridge: forward jobs to dispatcher
Bridge->>MeshNode: steal from peers if low
end
JavaWorker->>JNI: meshClusterInfo(handle)
JNI->>MeshNode: cluster_info()
MeshNode-->>JNI: JSON snapshot
JNI-->>JavaWorker: Optional<MeshClusterInfo>
Possibly related PRs
Suggested labels: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
sdks/java/src/main/java/org/byteveda/taskito/worker/MeshOptions.java (1)
96-185: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAdd range validation to remaining mesh option setters.
Only
port()validates its range;affinityWeight,virtualNodes,localBufferCapacity,maxStealBatch,stealThreshold, andstealRateLimitaccept arbitrary/negative values that are passed straight intotoConfigJson()and on to the native mesh node. The javadoc foraffinityWeightdocuments a[0.0, 1.0]contract that isn't enforced, and a non-positivevirtualNodescould be a source of native-side failures in the consistent-hash ring.🛡️ Example validation additions
public Builder affinityWeight(double affinityWeight) { + if (affinityWeight < 0.0 || affinityWeight > 1.0) { + throw new IllegalArgumentException("affinityWeight must be in [0.0, 1.0]"); + } this.affinityWeight = affinityWeight; return this; } ... public Builder virtualNodes(int virtualNodes) { + if (virtualNodes <= 0) { + throw new IllegalArgumentException("virtualNodes must be positive"); + } this.virtualNodes = virtualNodes; return this; }Please confirm whether the
taskito-meshcrate already guards these values server-side; if not, defensive validation here avoids surfacing a native-side crash as a confusing failure.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@sdks/java/src/main/java/org/byteveda/taskito/worker/MeshOptions.java` around lines 96 - 185, Add defensive range checks in MeshOptions.Builder for the remaining setters that currently accept unchecked values. Enforce the documented [0.0, 1.0] contract in affinityWeight(), and validate that virtualNodes, localBufferCapacity, maxStealBatch, stealThreshold, and stealRateLimit are non-negative or strictly positive as appropriate before storing them. Keep the checks alongside the existing port() validation in the Builder so invalid values are rejected before toConfigJson() serializes them and before the native mesh node sees them.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/taskito-java/src/mesh.rs`:
- Around line 30-35: MeshConfig parsing in the mesh setup is swallowing
malformed or incompatible JSON by falling back to defaults. In the mesh
initialization path where serde_json::from_str(config_json) is used, replace the
silent fallback with explicit error handling: log the parse failure with enough
context (for example, in the mesh scheduling enablement flow before the existing
log::info! call) and then choose an intentional fallback or abort path. Keep the
logging and config usage tied to the same MeshConfig parsing point so failures
are visible instead of silently running with default gossip_port and seeds.
- Around line 48-59: The background task in mesh.rs that uses
`tokio::join!(gossip, steal_server)` is discarding failures from the gossip and
steal-server workers. Update the async block that spawns these tasks to inspect
the `JoinError`/panic results from both tasks and log any failure with enough
context before exiting or propagating it. Use the existing `runtime.spawn`,
`gossip`, and `steal_server` symbols to locate the worker-lifetime task wrapper
and add failure handling there.
In `@crates/taskito-java/src/worker.rs`:
- Around line 100-123: The mesh setup in worker.rs is passing `capacity as u16`,
which can silently truncate when `channel_capacity` exceeds `u16::MAX`. Update
the `spawn_mesh` call path in `worker` to convert `capacity` safely by
saturating at `u16::MAX` (or otherwise clamping) before handing it to
`crate::mesh::spawn_mesh`, so the advertised threads/load-balancing value
remains correct.
---
Nitpick comments:
In `@sdks/java/src/main/java/org/byteveda/taskito/worker/MeshOptions.java`:
- Around line 96-185: Add defensive range checks in MeshOptions.Builder for the
remaining setters that currently accept unchecked values. Enforce the documented
[0.0, 1.0] contract in affinityWeight(), and validate that virtualNodes,
localBufferCapacity, maxStealBatch, stealThreshold, and stealRateLimit are
non-negative or strictly positive as appropriate before storing them. Keep the
checks alongside the existing port() validation in the Builder so invalid values
are rejected before toConfigJson() serializes them and before the native mesh
node sees them.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 023d2032-5dd8-40b9-8162-395cfebf36eb
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (13)
crates/taskito-java/Cargo.tomlcrates/taskito-java/src/convert.rscrates/taskito-java/src/lib.rscrates/taskito-java/src/mesh.rscrates/taskito-java/src/worker.rssdks/java/build.gradle.ktssdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.javasdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.javasdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.javasdks/java/src/main/java/org/byteveda/taskito/worker/MeshClusterInfo.javasdks/java/src/main/java/org/byteveda/taskito/worker/MeshOptions.javasdks/java/src/main/java/org/byteveda/taskito/worker/Worker.javasdks/java/src/test/java/org/byteveda/taskito/MeshWorkerTest.java
1 participant
What
Binds the decentralized mesh scheduler (
crates/taskito-mesh) into the Java SDK behind a newmeshcargo feature. A worker started withMeshOptionsinterposes aMeshNode(SWIM gossip + consistent-hash affinity + TCP work-stealing) between the scheduler and the dispatcher, and exposes a liveMeshClusterInfosnapshot. The DB stays the source of truth; scheduler and dispatcher stay mesh-unaware — same bridge pattern as the Python and Node shells.How
meshfeature (crates/taskito-java): optionaltaskito-meshdep;#[cfg(feature = "mesh")] mod mesh.spawn_meshbuilds the node, starts gossip + steal-server inside aruntime.enter()guard (they call the freetokio::spawn), and runs the bridge loop: scheduler output → affinity-sorted local deque (+ steal from busy peers) → dispatcher channel.WorkerOptions.mesh_config(raw JSON) selects the path; without it the plain scheduler loop is byte-identical.NativeWorker.meshClusterInforeturns a JSONClusterInfoornull;stopsignals the node so gossip/steal tasks exit instead of lingering.WorkerHandleholds the node for its lifetime.Worker.Builder.mesh(MeshOptions);MeshOptions(record-backed builder, port/seeds/buffer/intervals, range-validated);MeshClusterInfo;Worker.meshClusterInfo()→Optional.cargoBuildnow passes--features postgres,redis,workflows,mesh;Cargo.lockupdated for the new dep edge.Testing
MeshWorkerTest— a solo mesh worker runs a job end-to-end through the bridge and reportspeerCount == 0; a plain worker reports no cluster; out-of-range port rejected../gradlew buildon JDK 22: mesh (3), FFM (2), transport-parity (2) all ran (not skipped); Spotless/Checkstyle + suite green. Rustfmt/clippy --all-featuresclean; generic-crate leakage tripwire confirmstaskito-meshstays free of jni/pyo3/napi.Summary by CodeRabbit